package org.xq.softcup.mq.remoting.net.impl.netty.server;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.xq.softcup.mq.remoting.net.Server;
import org.xq.softcup.mq.remoting.net.impl.netty.codec.NettyDecoder;
import org.xq.softcup.mq.remoting.net.impl.netty.codec.NettyEncoder;
import org.xq.softcup.mq.remoting.net.params.XQRpcRequest;
import org.xq.softcup.mq.remoting.net.params.XQRpcResponse;
import org.xq.softcup.mq.remoting.provider.XQRpcProviderFactory;
import org.xq.softcup.mq.util.ThreadPoolUtil;

import java.util.concurrent.ThreadPoolExecutor;
@Slf4j
public class NettyServer extends Server {
    private Thread thread;

    public void start(final XQRpcProviderFactory xqRpcProviderFactory) throws Exception {
        this.thread = new Thread(new Runnable() {
            public void run() {
                final ThreadPoolExecutor serverHandlerPool = ThreadPoolUtil.makeServerThreadPool(NettyServer.class.getSimpleName());
                EventLoopGroup bossGroup = new NioEventLoopGroup();
                EventLoopGroup workerGroup = new NioEventLoopGroup();
                try {
                    ServerBootstrap bootstrap = new ServerBootstrap();
                    bootstrap.group(bossGroup, workerGroup)
                            .channel(NioServerSocketChannel.class)
                            .childHandler(new ChannelInitializer() {
                                @Override
                                protected void initChannel(Channel channel) throws Exception {
                                    channel.pipeline()
                                    .addLast( new ChannelHandler[]{new NettyDecoder(XQRpcRequest.class, xqRpcProviderFactory.getSerializer())})
                                    .addLast(new ChannelHandler[]{ new NettyEncoder(XQRpcResponse.class, xqRpcProviderFactory.getSerializer())})
                                    .addLast(new ChannelHandler[]{new NettyServerHandler(xqRpcProviderFactory, serverHandlerPool)});
                                }

                            }).childOption(ChannelOption.TCP_NODELAY, Boolean.valueOf(true)).childOption(ChannelOption.SO_KEEPALIVE, Boolean.valueOf(true));

                    ChannelFuture future = bootstrap.bind(xqRpcProviderFactory.getPort()).sync();

                    log.info("Netty服务启动成功, 类型 = {}, 端口号 = {}", NettyServer.class.getName(), Integer.valueOf(xqRpcProviderFactory.getPort()));
                    NettyServer.this.onStarted();


                    future.channel().closeFuture().sync();
                    return;
                } catch (Exception e) {
                    if ((e instanceof InterruptedException)) {
                        log.info("Netty服务停止.");
                    } else {
                        log.error("Netty服务异常.", e);
                    }
                } finally {
                    try {
                        serverHandlerPool.shutdown();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                    try {
                        workerGroup.shutdownGracefully();
                        bossGroup.shutdownGracefully();
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                    }
                }
            }
        });
        this.thread.setDaemon(true);
        this.thread.start();
    }

    public void stop()
            throws Exception {
        if ((this.thread != null) && (this.thread.isAlive())) {
            this.thread.interrupt();
        }
        onStoped();
        log.info("Netty服务销毁完成.");
    }
}
